package com.qyer.commons.concurrent;

import org.slf4j.Logger;

import java.util.concurrent.CountDownLatch;

/**
 * User: Z J Wu Date: 2016/07/28 Time: 09:53 Package: com.qyer.commons.concurrent
 */
public abstract class RecordProcessor<T extends PoisonPill<T>> implements Runnable {

  protected static Logger LOGGER;

  protected String name;

  protected GenericLinkedBlockingQueue<T> queue1;

  protected CountDownLatch stopSignal;

  public RecordProcessor(String name, GenericLinkedBlockingQueue<T> queue1,
                         CountDownLatch stopSignal) {
    this.name = name;
    this.queue1 = queue1;
    this.stopSignal = stopSignal;
  }

  public abstract void process(T t) throws Exception;

  public abstract void shutdown();

  @Override

  public void run() {
    try {
      while (true) {
        T t = queue1.take();
        if (t == null) {
          continue;
        }
        if (t.isPill()) {
          break;
        }
        try {
          process(t);
        } catch (Exception e) {
          LOGGER.warn("Current round of processing failed.", e);
        }
      }
      LOGGER.info("Taker({}) stopped.", name);
    } catch (InterruptedException e) {
      LOGGER.warn("Taker(" + name + ") interrupted.", e);
    } finally {
      shutdown();
      stopSignal.countDown();
    }
  }
}
